package com.niit.hjw.floorscalebycomm;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class FloorScaleByCommReducer extends Reducer<Text, Text, Text, Text> {
    private Text outv = new Text();
    private List<String> list = new ArrayList<>();
    private Map<String, Integer> map = new HashMap<>();
    private double commTotal;

    /**
     * reduce阶段的核心业务逻辑（根据相同小区名 进行统计楼层的占比，然后输出）
     *
     * @param key     小区名称, eg: 香水鸿门...
     * @param values  楼层, eg: 高层(共6层)...
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            map.put(value.toString(), map.getOrDefault(values.toString(), 0) + 1);
            commTotal++;
        }
        for (Map.Entry<String, Integer> entry : map.entrySet()) {
            list.add(entry.getKey());
            DecimalFormat df = new DecimalFormat("#.#####");// 保留5位小数
            String formattedNumber = df.format((entry.getValue() / commTotal));
            list.add(formattedNumber);
            outv.set(list.toString());
            context.write(key, outv);
            list.clear();
        }

        commTotal = 0;
        map.clear();
    }
}
